/**
 * Copyright © 2011 Creativor Studio.
 * About license information, please see LICENSE.txt.
 */
package org.creativor.rayson.client;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.creativor.rayson.api.CallFuture;
import org.creativor.rayson.common.RpcCallException;
import org.creativor.rayson.exception.CallExecutionException;
import org.creativor.rayson.exception.ReadInvocationException;
import org.creativor.rayson.exception.RpcException;
import org.creativor.rayson.exception.RpcExceptionFactory;
import org.creativor.rayson.exception.ServiceNotFoundException;
import org.creativor.rayson.exception.UnsupportedVersionException;

// TODO: Auto-generated Javadoc
/**
 * An implement of {@link CallFuture}.
 * 
 * @author Nick Zhang
 * @param <V>
 */
class CallFutureImpl<V> implements CallFuture<V>
{
	
	private AtomicBoolean cancelled;
	private Condition doneCondition;
	private Lock resultLock;
	private boolean done = false;
	private RpcCallException invocationException;
	private V result;
	private Class<?>[] exceptionTypes;
	
	/**
	 * Instantiates a new call future impl.
	 * 
	 * @param exceptionTypes
	 *            the exception types
	 */
	public CallFutureImpl(Class<?>[] exceptionTypes)
	{
		this.resultLock = new ReentrantLock();
		doneCondition = resultLock.newCondition();
		cancelled = new AtomicBoolean(false);
		this.exceptionTypes = exceptionTypes;
	}
	
	@Override
	public boolean cancel(boolean mayInterruptIfRunning)
	{
		if (isDone() || isCancelled()) return false;
		cancelled.set(true);
		if (mayInterruptIfRunning)
		{
			notifyDone();
		}
		return true;
	}
	
	@Override
	public V get() throws InterruptedException, CallExecutionException, RpcException
	{
		resultLock.lock();
		try
		{
			while (!isDone())
			{
				doneCondition.await();
			}
		} finally
		{
			resultLock.unlock();
		}
		if (invocationException != null)
		{
			throwException(invocationException);
		}
		return result;
	}
	
	private void throwException(RpcCallException callException) throws RpcException,
			CallExecutionException
	{
		//
		Throwable remoteException = callException.getCause();
		StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
		remoteException.setStackTrace(Arrays.copyOfRange(stackTraceElements,
				stackTraceElements.length - 1, stackTraceElements.length));
		
		if (callException.isInvokeException())
		{
			boolean declaredInvokeException = false;
			if (this.exceptionTypes != null)
			{
				for (Class exception : this.exceptionTypes)
				{
					if (exception.isAssignableFrom(remoteException.getClass()))
					{
						declaredInvokeException = true;
						break;
					}
				}
			}
			
			if (declaredInvokeException)
			{
				throw new CallExecutionException(remoteException);
			} else
				throw RpcExceptionFactory.createUndecleardException(remoteException);
		}
		
		if (remoteException instanceof ReadInvocationException)
			throw RpcExceptionFactory
					.createParameterException((ReadInvocationException) remoteException);
		
		if (remoteException instanceof UnsupportedVersionException)
			throw RpcExceptionFactory
					.createUnsupportedProxyVersion((UnsupportedVersionException) remoteException);
		
		if (remoteException instanceof ConnectionClosedException)
			
			throw RpcExceptionFactory
					.createNetWorkException((ConnectionClosedException) remoteException);
		
		if (remoteException instanceof ServiceNotFoundException)
			
			throw RpcExceptionFactory
					.createServiceNotFoundException((ServiceNotFoundException) remoteException);
		
		throw RpcExceptionFactory.createUndecleardException(remoteException);
		
	}
	
	/**
	 * Gets the.
	 * 
	 * @param timeout
	 *            the timeout
	 * @param unit
	 *            the unit
	 * @return the v
	 * @throws InterruptedException
	 *             the interrupted exception
	 * @throws RpcException
	 *             the rpc exception
	 * @throws TimeoutException
	 *             the timeout exception
	 * @throws CallExecutionException
	 *             the call execution exception
	 */
	@Override
	public V get(long timeout, TimeUnit unit) throws InterruptedException, RpcException,
			TimeoutException, CallExecutionException
	{
		long startTime = System.currentTimeMillis();
		long timeOutMillis = unit.toMillis(timeout);
		resultLock.lock();
		try
		{
			while (!isDone())
			{
				doneCondition.await(timeOutMillis, unit);
			}
		} finally
		{
			resultLock.unlock();
		}
		if (System.currentTimeMillis() - startTime > timeOutMillis) throw new TimeoutException();
		
		if (invocationException != null)
		{
			throwException(invocationException);
		}
		return result;
	}
	
	/**
	 * Checks if is cancelled.
	 * 
	 * @return true, if is cancelled
	 */
	@Override
	public boolean isCancelled()
	{
		return cancelled.get();
	}
	
	/**
	 * Checks if is done.
	 * 
	 * @return true, if is done
	 */
	@Override
	public boolean isDone()
	{
		if (isCancelled()) return true;
		return done;
	}
	
	private void notifyDone()
	{
		resultLock.lock();
		try
		{
			doneCondition.signalAll();
		} finally
		{
			resultLock.unlock();
		}
	}
	
	/**
	 * Sets the.
	 * 
	 * @param v
	 *            the v
	 */
	public void set(V v)
	{
		this.result = v;
		this.done = true;
		notifyDone();
	}
	
	/**
	 * Sets the exception.
	 * 
	 * @param t
	 *            the new exception
	 */
	public void setException(RpcCallException t)
	{
		this.invocationException = t;
		this.done = true;
		notifyDone();
	}
	
}
